package cn._51doit.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @create: 2021-10-21 20:13
 * @author: 今晚打脑斧先森
 * @program: RestartStrategyDemo3
 * @Description:
 *    设置Flink程序的重启策略
 *    实时计算程序，如果出现了异常，最好要可以自我恢复，可以保证程序继续运行。
 *    无限重启
 **/
public class RestartStrategyDemo3 {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        //如果开启了checkpoint,那么job的默认重启策略为无限重启(Integer.MAX_VALUE)
        env.enableCheckpointing(Integer.MAX_VALUE);
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> res = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] fields = value.split(",");
                String word = fields[0];
                if (word.startsWith("error")) {
                    throw new RuntimeException("亲爱的今晚打脑斧先森 : 数据输入有问题哦~~");
                }
                int count = Integer.parseInt(fields[1]);
                return Tuple2.of(word, count);
            }
        });

        res.print();
        env.execute();
    }
}
